package com.rd.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.rabbitmq.client.Channel;
import com.rd.dto.Result;
import com.rd.entity.SeckillVoucher;
import com.rd.entity.VoucherOrder;
import com.rd.mapper.VoucherOrderMapper;
import com.rd.service.SecKillVoucherService;
import com.rd.service.VoucherOrderService;
import com.rd.utils.RedisIdWorker;
import com.rd.utils.SimpleRedisLock;
import com.rd.utils.UserHolder;
import lombok.extern.log4j.Log4j2;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.rd.utils.SnowflakeIdWorker.getUUID;

@Log4j2
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements VoucherOrderService {

    private static final String SEC_QUEUE = "seckillQueue";
    private static final String SEC_EXCHANGE = "seckillExchange";

    @Resource
    SecKillVoucherService secKillVoucherService;

    @Resource
    RedisIdWorker redisIdWorker;

    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Resource
    RedissonClient redissonClient;


//    @Resource
//    RabbitTemplate rabbitTemplate;

    // 创建堵塞队列
    BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024);

    ExecutorService es = Executors.newSingleThreadExecutor();

    //@PostConstruct作用：spring容器初始化的时候执行该方法
    //非静态的void()方法
    @PostConstruct
    public void initTask(){
        es.submit(new VoucherOrderHandler());
    }

    private class VoucherOrderHandler implements Runnable{
        @Override
        public void run() {
            final String queueName = "stream.orders";
            while (true){
                // 基于java 堵塞队列
                try {
                    VoucherOrder voucherOrder = orderTasks.take();
                    createVoucherOrder1(voucherOrder);
                } catch (Exception e) {
                    System.out.println("订单处理异常");
                    e.printStackTrace();
                }

                try {
                    //使用redis的stream消息队列实现异步下单
                    // 获取消息队列中的订单信息
                    //命令：XREADGROUP 	GROUP 	group 	consumer
                    // [COUNT count] 	[BLOCK milliseconds] [NOACK]	 STREANS	key	 [key ...]	ID 	[ID ...]
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().
                            read(
                                    Consumer.from("g1", "c1"),
                                    // 堵塞2秒等待
                                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                                    StreamOffset.create(queueName, ReadOffset.lastConsumed())
                            );


                    // 判断订单信息是否为空
                    if (ObjectUtils.isEmpty(list)){
                        continue;   // 没有消息，下一次循环
                    }
                    // 解析处理消息
                    handlerMsg(queueName, list);
                } catch (Exception e) {
                    System.out.println("订单处理异常");
                    e.printStackTrace();
                    // 处理未确认消息
                    handlerPendingList();
                }
            }
        }

        private void handlerPendingList() {
            final String queueName = "stream.orders";
            while (true) {
                try {
                    // 1、获取PendingList消息队列中的订单信息
                    //命令：XREADGROUP 	GROUP 	group 	consumer
                    // [COUNT count] 	[BLOCK milliseconds] [NOACK]	 STREANS	key	 [key ...]	ID 	[ID ...]
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().
                            read(
                                    Consumer.from("g1", "c1"),
                                    StreamReadOptions.empty().count(1),
                                    StreamOffset.create(queueName, ReadOffset.from("0"))
                            );


                    // 2、判断订单信息是否为空
                    if (list.isEmpty() || list == null){
                        break;   // 没有消息,直接退出循环
                    }
                    handlerMsg(queueName, list);
                } catch (Exception e) {
                    System.out.println("订单处理异常");
                    e.printStackTrace();
                    try {
                        Thread.sleep(300);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }

        private void handlerMsg(String queueName, List<MapRecord<String, Object, Object>> list) {
            // 解析消息
            MapRecord<String, Object, Object> entries = list.get(0);
            Map<Object, Object> value = entries.getValue();
            // map 转为 VoucherOrder
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

//            System.out.println("订单信息："+voucherOrder);

            // 3、创建订单
            createVoucherOrder1(voucherOrder);

            //确认消息  命令  XACK  key   group   id
            stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", entries.getId());
        }


    }

    static DefaultRedisScript SECKILL_SCRIPT;

    static {
        SECKILL_SCRIPT = new DefaultRedisScript();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }


    private boolean createVoucherOrder2(VoucherOrder voucherOrder) {

        boolean flag = false;
        // 每个用户只能抢一张优惠券
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();

        // 创建锁对象
        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
        // 尝试获取锁
        boolean lock = redisLock.tryLock();
        if (!lock){
            System.out.println("不允许重复下单！！");
            return false;
        }
        try {
            int count = query()
                    .eq("user_id", userId)
                    .eq("voucher_id", voucherId)
                    .count();
            if (count > 0) {
                System.out.println("您已经抢过了!");
                return false;
            }
            //5、扣除库存
            boolean res = secKillVoucherService.update()
                    // set stock = stock - 1
                    .setSql("stock = stock - 1")
                    // where voucher_id = ? and stock > 0
                    .eq("voucher_id", voucherId)
                    .gt("stock", 0)
                    .update();
            // 扣除失败
            if (!res) {
                System.out.println("库存不足！！");
                return false;
            }
            //6、创建订单
//            VoucherOrder voucherOrder = new VoucherOrder();
//            long orderId = redisIdWorker.nextId("order");
//            voucherOrder.setId(orderId);
//            voucherOrder.setUserId(userId);
//            voucherOrder.setVoucherId(voucherId);
            // 7、保存到数据库
            flag = save(voucherOrder);

        } catch (Exception e) {

            e.printStackTrace();
        } finally {
            redisLock.unlock(); // 释放锁
        }
        return flag;
    }
    private void createVoucherOrder1(VoucherOrder voucherOrder) {
        // 每个用户只能抢一张优惠券
        Long userId = voucherOrder.getUserId();
        Long voucherId = voucherOrder.getVoucherId();

        // 创建锁对象
        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
        // 尝试获取锁
        boolean lock = redisLock.tryLock();
        if (!lock){
            System.out.println("不允许重复下单！！");
            return;
        }
        try {
            int count = query()
                    .eq("user_id", userId)
                    .eq("voucher_id", voucherId)
                    .count();
            if (count > 0) {
                System.out.println("您已经抢过了!");
                return;
            }
            //5、扣除库存
            boolean res = secKillVoucherService.update()
                    // set stock = stock - 1
                    .setSql("stock = stock - 1")
                    // where voucher_id = ? and stock > 0
                    .eq("voucher_id", voucherId).gt("stock", 0)
                    .update();
            // 扣除失败
            if (!res) {
                System.out.println("库存不足！！");
                return;
            }
            //6、创建订单
//            VoucherOrder voucherOrder = new VoucherOrder();
//            long orderId = redisIdWorker.nextId("order");
//            voucherOrder.setId(orderId);
//            voucherOrder.setUserId(userId);
//            voucherOrder.setVoucherId(voucherId);
            // 7、保存到数据库
            save(voucherOrder);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            redisLock.unlock(); // 释放锁
        }

    }


    //TODO 监听队列方法
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = SEC_QUEUE,
                    durable = "true"),
            exchange = @Exchange(value = SEC_EXCHANGE,
                    ignoreDeclarationExceptions = "true"),
            key = "seckill")  // routing key
    )
    @RabbitHandler
    public void receiveMsg(@Payload VoucherOrder voucherOrder,@Headers Map<String, Object> heads,
                           Channel channel) throws IOException {
        log.info("------监听队列中--------");
        //使用雪花算法生成id
        voucherOrder.setId(getUUID());
        voucherOrder.setCreateTime(LocalDateTime.now());
        // 创建订单
        boolean b = createVoucherOrder2(voucherOrder);
        Long deliverTag = (Long) heads.get(AmqpHeaders.DELIVERY_TAG);//唯一标识ID
        if (b){
            System.out.println("订单生成成功");
            //手动ack确认
            // 默认自动ack
            // 需要在配置文件中配置acknowledge-mode为：manual
            channel.basicAck(deliverTag, false);
        }else{
            System.out.println("订单生成失败！！");

            //拒绝消息
            //basicReject(msgId,requeue)
            //requeue为true，消息重回队列，并通过轮循机制推送给消费者
            //requeue为false，将消息丢弃
            channel.basicReject(deliverTag,false);

            //可以拒绝多条消息
            //channel.basicNack(msgId,multi,requeue)
        }

    }
    /**
     * 异步秒杀下单
     */
    public Result asyncSeckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        long orderId = redisIdWorker.nextId("order");
        Long res = (Long) stringRedisTemplate.execute(SECKILL_SCRIPT,
                // 没有key
                Collections.emptyList(),
                voucherId.toString(),
                userId.toString(),
                String.valueOf(orderId));

        Map<String,Object> map = new HashMap<>();
        map.put("voucherId",voucherId.toString());
        map.put("userId",userId.toString());

        if (res != 0){
            return Result.fail(res == 1 ? "库存不足!!":"不允许重复下单!!");
        }

        // 发送消息消息队列
//        rabbitTemplate.convertAndSend(SEC_EXCHANGE,SEC_QUEUE,map);

        // java堵塞队列
        //创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        //保存到堵塞队列
        orderTasks.add(voucherOrder);
        return Result.ok(orderId);
    }

    /**
     * 同步秒杀下单
     */
    @Override
    public Result syncSeckillVoucher(Long voucherId) {

        //查询优惠券
        SeckillVoucher voucher = secKillVoucherService.getById(voucherId);

        //判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())){
            return Result.fail("秒杀尚未开始！！");
        }
        //判断秒杀是否结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())){
            return Result.fail("秒杀已经结束！！");
        }
        //判断库存是否充足
        if (voucher.getStock() < 1) {
            return Result.fail("库存不足！！");
        }

        /** 单体锁，集群环境下不适用
         * toString方法会new一个新对象，因此需要使用intern方法
         * intern方法如果字符串常量池中已经存在一个等于此String对象的字符串
         * 就直接从常量池中返回这个字符串对象的引用
         */
//        synchronized (userId.toString().intern()) {
//            /**
//             * 要想事务生效，必须使用代理对象来调用
//             */
//            // 拿到当前对象的代理对象
//            VoucherOrderService voucherOrderService = (VoucherOrderService) AopContext.currentProxy();
//            return voucherOrderService.createVoucherOrder(voucherId);
//        }
        /**
         * 分布式锁
         */
        return createVoucherOrderByRC(voucherId);
    }

    // 使用RedissonClient 获取锁，优点：可重入，可重试，超时续约
    @Transactional
    public Result createVoucherOrderByRC(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        // 创建锁对象
        RLock redisLock = redissonClient.getLock("lock:order:" + userId);
        // 尝试获取锁
        boolean lock = redisLock.tryLock();
        if (!lock){
            return Result.fail("不允许重复下单！！");
        }
        try {
             /**
             * 要想事务生效，必须使用代理对象来调用
             */
            // 拿到当前对象的代理对象
            VoucherOrderService voucherOrderService = (VoucherOrderService) AopContext.currentProxy();
            return voucherOrderService.createVoucherOrder(voucherId);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            redisLock.unlock(); // 释放锁
        }
        return Result.fail("抢购失败！！");
    }
    // 使用setnx命令获取锁,缺点：不可重入、不可重试、超时释放
    @Transactional
    public Result createVoucherOrderByMyLock(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        SimpleRedisLock simpleRedisLock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        boolean lock = simpleRedisLock.tryLock(60);
        if (!lock){
            return Result.fail("不允许重复下单！！");
        }
        try {

            /**
             * 要想事务生效，必须使用代理对象来调用
             */
            // 拿到当前对象的代理对象
            VoucherOrderService voucherOrderService = (VoucherOrderService) AopContext.currentProxy();
            return voucherOrderService.createVoucherOrder(voucherId);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            simpleRedisLock.unlockByLuaScript();
        }
        return Result.fail("抢购失败！！");
    }

    @Override
    @Transactional
    public Result createVoucherOrder(Long voucherId) {
        // 从ThreadLocal中获取用户id
        Long userId = UserHolder.getUser().getId();
        int count = query()
                .eq("user_id", userId)
                .eq("voucher_id", voucherId)
                .count();
        if (count > 0) {
            return Result.fail("您已经抢过了");
        }
        //扣除库存
        boolean res = secKillVoucherService.update()
                // set stock = stock - 1
                .setSql("stock = stock - 1")
                // where voucher_id = ? and stock > 0
                .eq("voucher_id", voucherId).gt("stock", 0)
                .update();
        if (!res) {
            return Result.fail("库存不足！！");
        }
        //创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);
        return Result.ok(orderId);
    }
}
